package com.atguigu.flink.timer;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/23
 *
 * 需求：监控水位传感器的水位值，如果水位值在5s之内(event time)连续上升，则报警。
 *
 *      ①第一条传感器的数据到达是，注册一个 基于事件时间的 5s后触发的定时器。
 *      ②后续的数据到达后，和前一条进行vc对比，如果上升，不处理
 *                      如果vc下降，取消定时器
 *      ③定时器不管是取消后，还是被触发后，一切清零，从新开始
 *
 *
 *      测试时，先只用一种key!
 */
public class Demo4_TimerExec
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //水印策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()   //目前的场景数据都是有序的
                                                      //如何产生水印
                                                      .withTimestampAssigner((e, r) -> e.getTs());
        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .keyBy(WaterSensor::getId)
           .process(new KeyedProcessFunction<String, WaterSensor, String>()
           {
               @Override
               public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("5s之内水位连续上升...");
                        //从新开始
                        ifFirst = true;
               }

               //声明一个变量，保存定时器的绝对时间
               long time = 0;

               //声明一个变量，标识当前数据是否是第一条
               boolean ifFirst = true;
               //声明一个变量，保存上一条的vc
               int lastVc = 0;

               @Override
               public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                   TimerService timerService = ctx.timerService();

                   if (ifFirst) {
                       //第一条数据来
                       time = ctx.timestamp() + 5000;
                       System.out.println("注册了定时器:"+time);
                       timerService.registerEventTimeTimer(time);
                       //修改标记
                       ifFirst = false;
                   }else{
                       //不是第一条，对比上一条的vc
                       if (value.getVc() < lastVc){
                           //下降
                           System.out.println("删除了定时器:"+time);
                           timerService.deleteEventTimeTimer(time);
                           //从新开始
                           ifFirst = true;
                       }

                   }

                   //每一条来，都要把自己的vc保存下来，给下一条用
                   lastVc = value.getVc();

                   out.collect(value.toString());
               }
           })
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
